package netty;

import util.ByteBufferUtil;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadsServer {
  public static void main(String[] args) {
    try (ServerSocketChannel server = ServerSocketChannel.open()) {
      // 当前线程为Boss线程
      Thread.currentThread().setName("Boss");
      server.bind(new InetSocketAddress(8080));
      // 负责轮询Accept事件的Selector
      Selector boss = Selector.open();
      server.configureBlocking(false);
      server.register(boss, SelectionKey.OP_ACCEPT);
      // 创建固定数量的Worker
      Worker[] workers = new Worker[4];
      // 用于负载均衡的原子整数
      AtomicInteger robin = new AtomicInteger(0);
      for(int i = 0; i < workers.length; i++) {
        workers[i] = new Worker("worker-"+i);
      }
      while (true) {
        boss.select();
        Set<SelectionKey> selectionKeys = boss.selectedKeys();
        Iterator<SelectionKey> iterator = selectionKeys.iterator();
        while (iterator.hasNext()) {
          SelectionKey key = iterator.next();
          iterator.remove();
          // BossSelector负责Accept事件
          if (key.isAcceptable()) {
            // 建立连接
            SocketChannel socket = server.accept();
            System.out.println("connected...");
            socket.configureBlocking(false);
            // socket注册到Worker的Selector中
            System.out.println("before read...");
            // 负载均衡，轮询分配Worker
            workers[robin.getAndIncrement()% workers.length].register(socket);
            System.out.println("after read...");
          }
        }
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  static class Worker implements Runnable {
    private Thread thread;
    private volatile Selector selector;
    private String name;
    private volatile boolean started = false;
    /**
     * 同步队列，用于Boss线程与Worker线程之间的通信
     */
    private ConcurrentLinkedQueue<Runnable> queue;

    public Worker(String name) {
      this.name = name;
    }

    public void register(final SocketChannel socket) throws IOException {
      // 只启动一次
      if (!started) {
        thread = new Thread(this, name);
        selector = Selector.open();
        queue = new ConcurrentLinkedQueue<>();
        thread.start();
        started = true;
      }

      // 向同步队列中添加SocketChannel的注册事件
      // 在Worker线程中执行注册事件
      queue.add(new Runnable() {
        @Override
        public void run() {
          try {
            socket.register(selector, SelectionKey.OP_READ);
          } catch (IOException e) {
            e.printStackTrace();
          }
        }
      });
      // 唤醒被阻塞的Selector
      // select类似LockSupport中的park，wakeup的原理类似LockSupport中的unpark
      selector.wakeup();
    }

    @Override
    public void run() {
      while (true) {
        try {
          selector.select();
          // 通过同步队列获得任务并运行
          Runnable task = queue.poll();
          if (task != null) {
            // 获得任务，执行注册操作
            task.run();
          }
          Set<SelectionKey> selectionKeys = selector.selectedKeys();
          Iterator<SelectionKey> iterator = selectionKeys.iterator();
          while(iterator.hasNext()) {
            SelectionKey key = iterator.next();
            iterator.remove();
            // Worker只负责Read事件
            if (key.isReadable()) {
              // 简化处理，省略细节
              SocketChannel socket = (SocketChannel) key.channel();
              ByteBuffer buffer = ByteBuffer.allocate(16);
              socket.read(buffer);
              buffer.flip();
              ByteBufferUtil.debugAll(buffer);
            }
          }
        } catch (IOException e) {
          e.printStackTrace();
        }
      }
    }
  }
}
